package com.herenit.order_model.utils.kafka;

import com.herenit.order_model.enums.ExceptionEnums;
import com.herenit.order_model.exception.MyException;
import com.herenit.order_model.model.Message;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.*;

@Component
public class KafkaConsumerUtils {

    @Autowired
    private KafkaConsumer<String, Object> consumer;

    private static String PRIOR = "topic_prior_";

    private static String NORMAL = "topic_normal_";

    private Message rtnMessage;

    /**
     * 根据机构Code 去对应的topic中拉取消息
     *
     * @param institution
     * @return
     */
    public Message consume(String institution) {
        // 订阅优先队列
        consumer.subscribe(Arrays.asList(PRIOR + institution));
        ConsumerRecords<String, Object> records;
        try {
            // 尝试从优先队列中拉取消息
            records = consumer.poll(Duration.ofMillis(1000));
            // 为空 则订阅一般队列
            if (records.isEmpty()) {
                // 从一般队列中拉取消息
                consumer.subscribe(Arrays.asList(NORMAL + institution));
                records = consumer.poll(Duration.ofMillis(1000));
            }
        } catch (Exception e) {
            throw new MyException(ExceptionEnums.KAFKA_POLL_ERROR);
        }
        // 消息队列全为空
        if (records.isEmpty()) {
            return new Message("0", "消息队列为空", records);
        }
        // 模拟消费失败 随机发生 消费失败 code 为 0
        int random = new Random().nextInt(100);
        if (random < 50) {
            rtnMessage = new Message("0", "消费异常,加入优先队列", records);
            if (random > 10) {
                rtnMessage.setMessage("返回异常,加入优先队列");
            }
            if (random > 25) {
                rtnMessage.setMessage("处理异常,加入优先队列");
            }
            if (random > 40) {
                rtnMessage.setMessage("其他异常,加入优先队列");
            }
            return rtnMessage;
        }
        rtnMessage = new Message("200", "消费成功", records);
        return rtnMessage;
    }

}
